﻿/*----------------------------------------------------------------------------------------
*
*  Copyright 2019, Gao Hai Hui, <fromzeropoint@126.com>.  All rights reserved.
*  https://gitee.com/helloworldghh/xoskit.git
*  Use of this source code is governed by a MIT license
*  that can be found in the License file.
*
----------------------------------------------------------------------------------------*/
#include "../import/head.h"
#include "../global/head.h"
#include "../macro/head.h"
#include "aio_epoll_notify.h"
#include "aio_helper.h"
#include "aio_impl.h"
#include "aio_thread.h"

/////////////////////////////////////////////////////////////////////////////////////////
// 
namespace xos_aio
{
    aio_thread::aio_thread( aio_impl * pAioImpl ) : m_pAioImpl( pAioImpl )
    {
        m_bQuit = false;
        m_pThread = 0;
    }

    aio_thread::~aio_thread()
    {
    }

    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // 

    int aio_thread::init()
    {
        int nRet = 0;

        if( 0 == nRet )
        {
            nRet = g_pSysData->m_pContainer->xos()->create(xos::i_xos::XOS_OBJ_THREAD, ( void ** )&m_pThread);
        }

        if( 0 == nRet )
        {
            m_pThread->create( aio_thread::thread, this );
        }

        return nRet;
    }

    int aio_thread::term()
    {
        int nRet = 0;

        if( m_pThread )
        {
            m_bQuit = true;
            m_pAioImpl->m_pEpollNotify->set_event();
            m_pThread->wait();
            m_pThread->release();
            m_pThread = 0;
        }

        m_pAioImpl = 0;

        return nRet;
    }

    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
    // 

    void * aio_thread::thread( void * lpData )
    {
        aio_thread & obj = *( aio_thread * )lpData;
        aio_impl * lpAioImpl = obj.m_pAioImpl;

        while( !obj.m_bQuit )
        {
            int nNum = epoll_wait( lpAioImpl->m_hEpoll, obj.m_event, sizeof( obj.m_event ) / sizeof( obj.m_event[0] ), lpAioImpl->m_nTimeout );
            if( nNum > 0 )
            {
                obj.run( nNum );
            }
        }

        return 0;
    }

    int aio_thread::run( int nNum )
    {
        using namespace std;
        int nRet = 0;

        aio_data::LIST & data_list = m_pAioImpl->m_return_list;

        if( 0 == nRet )
        {
            data_list.put_back_to_pool( true );
        }

        for( int i = 0; ( 0 == nRet ) && ( i < nNum ); ++i )
        {
            helper_run( m_event[i] );
        }

        for( aio_data::ITER i = data_list.begin(); ( 0 == nRet ) && ( i != data_list.end() ); ++i )
        {
            aio_data * p = *i;
            xos_box::i_msg * pMsg = p->m_pMsg;
            aio_key * pKey = (aio_key *)pMsg->get_void( 1, 0 );
            pKey->finish_request( m_pAioImpl );
            if( aio_key::KEY_STATE_CLOSING == pKey->m_eState )
            {
                pMsg->set_void( 1, 0 );
                pKey->release();
                pKey = 0;
            }
            m_pAioImpl->notify( pMsg );
            p->m_pMsg = 0;
            pMsg = 0;
        }

        data_list.put_back_to_pool( true );

        return nRet;
    }

    ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

    int aio_thread::helper_run( epoll_event & e )
    {
        int ret = 0;

        if( e.data.ptr == m_pAioImpl->m_pEpollNotify )
        {
            m_pAioImpl->m_pEpollNotify->reset();
            ret = proc_user_request();
        }
        else
        {
            ret = proc_epoll_event( e );
        }

        return ret;
    }

    int aio_thread::proc_user_request()
    {
        int ret = 0;

        aio_data::LIST & data_list = m_pAioImpl->m_request_list;

        while( 1 )
        {
			aio_data * p = 0;
            data_list.lock();
			if( data_list.size() > 0 )
			{
				p = data_list.front();
				data_list.pop_front();
			}
            data_list.un_lock();

            if( !p )
            {
                break;
            }

            xos_box::i_msg * pMsg = p->m_pMsg;
            
            p->m_pMsg = 0;
            p->release();
            p = 0;

            aio_helper tool( m_pAioImpl );
            tool.user_proc( pMsg );
            pMsg = 0;
        }

        return ret;
    }

    int aio_thread::proc_epoll_event( epoll_event & e )
    {
        using namespace std;
        int ret = 0;

        xos::xos_u64 u = ( xos::xos_u64 )e.data.ptr;
        int nFlags = e.events;

        aio_key * lpKey = ( aio_key *) ( u & ( xos::xos_u64 ) ~1 );
        int nInstance = ( int )( u & 1 );
        lpKey->m_nLastInstance = nInstance;

        if( ( 0 == ret ) && ( nInstance != lpKey->m_nInstance ) )
        {
            cout << "stale event : instance" << endl;
            ret = 1;
        }

        if( ( 0 == ret ) && ( aio_key::KEY_STATE_NONE == lpKey->m_eState ) )
        {
            cout << "key state wrong" << endl;
            ret = 2;
        }

        if( 0 == ret )
        {
            aio_helper tool( m_pAioImpl );
            ret = tool.epoll_proc( lpKey, nFlags );
        }

        return ret;
    }

} // xos_aio
